package com.rd.service.impl;

import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.rd.dto.Result;
import com.rd.entity.Blog;
import com.rd.entity.RedisData;
import com.rd.entity.Shop;
import com.rd.mapper.ShopMapper;
import com.rd.service.ShopService;
import com.rd.utils.RedisCacheUtils;
import com.rd.utils.SystemConstants;
import lombok.NonNull;
import lombok.val;
import org.redisson.api.RBloomFilter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.geo.Distance;
import org.springframework.data.geo.GeoResult;
import org.springframework.data.geo.GeoResults;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.data.redis.domain.geo.GeoReference;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static com.rd.utils.RedisConstants.SHOP_GEO_KEY;

@Service
public class ShopServiceImpl extends ServiceImpl<ShopMapper, Shop> implements ShopService {

    final static String CACHE_SHOP_KEY = "cache:shop:";

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    RedisCacheUtils redisCacheUtils;


    @Autowired
    RBloomFilter bloomFilter;

    /**
     * 初始化布隆过滤器
     */
    @PostConstruct
    public void initBloomFilter(){

        List<Long> ids = list().stream()
                            .map(Shop::getId)
                            .collect(Collectors.toList());
//        // 设置布隆过滤器的误判率
//        bloomFilter.tryInit(ids.size(),0.01);
        // 将所有店铺信息添加到布隆过滤器
        bloomFilter.add(ids);
    }

    // 创建一个线程池 线程数目为10
    private static final ExecutorService es = Executors.newFixedThreadPool(10);


    //TODO 加互斥锁并添加过期时间 boot2.1版本以上才支持
    private Boolean setIfAbsent(String key){
        Boolean b = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
        return BooleanUtil.isTrue(b);
    }
    //TODO boot2.1以下 使用lua脚本设置锁保持原子性
    private Boolean tryGetLockByLua(String key){
        // KEYS[1] 用来表示在redis 中用作键值的参数占位，
        // 主要用來传递在redis 中用作keys值的参数
        // ARGV[1] 用来表示在redis 中用作参数的占位，
        // 主要用来传递在redis中用做value值的参数。
        String newSetIfAbsentScriptStr = " if 1 == redis.call('setnx',KEYS[1],ARGV[1]) then" +
                " return 1;" +
                " else" +
                " return 0;" +
                " end;";
        //创建 redis脚本对象
        RedisScript<Boolean> newSetIfAbsentScript = new DefaultRedisScript<>(newSetIfAbsentScriptStr,Boolean.class);
        List<String> keys = new ArrayList<>();
        keys.add(key);  // key
        Object[] values = {"1"};  // value
        // 执行脚本
        Boolean b = stringRedisTemplate.execute(newSetIfAbsentScript, keys, values);
        return BooleanUtil.isTrue(b);

    }
    // 删除锁
    private void unLock(String key){
        stringRedisTemplate.delete(key);
    }


    public void saveShopToRedis(Long id,Long expireSeconds) throws InterruptedException {
        // 根据id查询数据库
        Shop shop = getById(id);
        // 封装 逻辑过期时间
        RedisData data = new RedisData();
        data.setData(shop);
        data.setExpireTime(LocalDateTime.now().plusSeconds(expireSeconds));
        //写入缓存
        stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY+id,JSONUtil.toJsonStr(data));
    }

    /**
     * 逻辑过期解决缓存击穿
     */
    private Shop queryWithLogicalExpire(Long id){
        String key = CACHE_SHOP_KEY + id;
        // 查询缓存
        String shopJson = stringRedisTemplate.opsForValue().get(key);
        // 缓存命中，先将json反序列化成对象
        RedisData  data = JSONUtil.toBean(shopJson, RedisData.class);
        JSONObject jsonData = (JSONObject) data.getData();
        Shop notExpireShop = JSONUtil.toBean(jsonData, Shop.class);
        LocalDateTime expireTime = data.getExpireTime();
        // 过期时间在当前时间之后，数据未过期，直接返回
        if (expireTime.isAfter(LocalDateTime.now())){
            return notExpireShop;
        }

        // 已过期，缓存重建
        String lockKey = "lock:shop:"+id;
        //获得互斥锁
        Boolean lock = setIfAbsent(lockKey);
        //判断是否成功获得锁
        if(lock){
            //获取锁成功，开启独立线程，缓存重建
            es.submit(()->{
                try {
                    //缓存重建
                    this.saveShopToRedis(id, 20L);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    // 释放锁
                    unLock(lockKey);
                }
            });
        }
        //锁获取失败，返回旧的商铺信息
        return notExpireShop;

    }

    /**
     * 布隆过滤器解决缓存穿透
     */
    public Shop queryWithPassThroughByBf(Long id){
        boolean b = bloomFilter.contains(id);
        String key = CACHE_SHOP_KEY + id;
        if (b){
            // 查询缓存
            String shopJson = stringRedisTemplate.opsForValue().get(key);
            if (StrUtil.isNotBlank(shopJson)){
                Shop shop = JSONUtil.toBean(shopJson, Shop.class);
                return shop;
            }
            // ""空值不是null
            if (shopJson != null){
                return null;
            }
            // 查询数据库
            Shop shop = getById(id);
            if (shop == null){
                // 若数据不存在，写入空值来应对缓存穿透
                stringRedisTemplate.opsForValue().set(key, "",2, TimeUnit.MINUTES);
                return null;
            }
            // 保存缓存到redis
            stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop),10, TimeUnit.MINUTES);
            return shop;
        }
        return null;
    }
    /**
     * 缓存空值解决缓存穿透
     */
    private Shop queryWithPassThrough(Long id){
        String key = CACHE_SHOP_KEY + id;
        // 查询缓存
        String shopJson = stringRedisTemplate.opsForValue().get(key);
        if (StrUtil.isNotBlank(shopJson)){
            Shop shop = JSONUtil.toBean(shopJson, Shop.class);
            return shop;
        }else if (shopJson != null){
            return null;
        }
        // 查询数据库
        Shop shop = getById(id);
        if (shop == null){
            // 若数据不存在，写入空值来应对缓存穿透
            stringRedisTemplate.opsForValue().set(key, "",2, TimeUnit.MINUTES);
            return null;
        }
        // 保存缓存到redis
        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop),10, TimeUnit.MINUTES);
        return shop;
    }

    //TODO 互斥锁解决缓存击穿
    private Shop queryWithMutex(Long id){
        String key = CACHE_SHOP_KEY + id;
        // 查询缓存
        String shopJson = stringRedisTemplate.opsForValue().get(key);
        // 不是null,说明有缓存数据
        if (StrUtil.isNotBlank(shopJson)){
            Shop shop = JSONUtil.toBean(shopJson, Shop.class);
            return shop;
        // 不是空值但可能是空字符串 或 换行符制表符等
        }else if (shopJson != null){
            return null;
        }
        //实现缓存重建(查询数据库)
        Shop shop = null;
        String lockKey = "lock:shop:"+id;
        try {
            //获取互斥锁
            Boolean flag = tryGetLockByLua(lockKey);
            //判断锁是否获取成功
            if ( !flag ){
                //锁获取失败，则休眠并重试
                Thread.sleep(50);
                return queryWithMutex(id);
            }
            //锁获取成功 查询数据库
            shop = getById(id);

            // 模拟延迟
            Thread.sleep(200);

            // 不存在，返回错误信息
            if (shop == null){
                // 若数据不存在，写入空值；来应对缓存穿透
                stringRedisTemplate.opsForValue().set(key, "",2, TimeUnit.MINUTES);
                return null;
            }
            //保存缓存到redis
            stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop),10, TimeUnit.MINUTES);

        } catch (InterruptedException e) {
            throw new RuntimeException();
        }finally {
            // 7.释放互斥锁
            unLock(lockKey);
        }
        return shop;
    }

    @Override
    public Result queryShopById(Long id) {
          //解决缓存穿透
//        Shop shop = queryWithPassThrough(id);
        //互斥锁解决缓存击穿
//        Shop shop = queryWithMutex(id);

        //逻辑过期锁解决缓存击穿
//        Shop shop = queryWithLogicalExpire(id);

        // 解决缓存穿透
//      Shop shop = redisCacheUtils.queryWithPassThrough(CACHE_SHOP_KEY, id, 10L, TimeUnit.MINUTES, Shop.class, this::getById);

        // 布隆过滤器解决缓存穿透
        Shop shop = queryWithPassThroughByBf(id);

//        // 使用逻辑过期解决缓存击穿
//        Shop shop = redisCacheUtils.queryWithLogicalExpire(CACHE_SHOP_KEY, id, 10L, TimeUnit.MINUTES, Shop.class, this::getById);
        if (shop == null){
            return Result.fail("店铺不存在");
        }
        return Result.ok(shop);
    }

    @Override
    public Result updateShop(Shop shop) {

        // 1、更新数据库
        val id = shop.getId();
        String key = "cache:shop:" + id;
        if (id == null){
            return Result.fail("店铺id不能为空");
        }
        updateById(shop);
        // 2、删除缓存
        stringRedisTemplate.delete(key);
        return Result.ok();
    }

    // 返回附近商户距离并分页及排序
    @Override
    public Result queryShopByType(Integer typeId, Integer current, Double x, Double y) {

        // 判断是否需要根据坐标查询
        if (x==null || y==null){
            Page page = new Page<>(current, SystemConstants.DEFAULT_PAGE_SIZE);
            query().eq("type_id", typeId).page(page);
            return Result.ok(page.getRecords());
        }
        //计算分页参数
        int from = (current - 1) * SystemConstants.DEFAULT_PAGE_SIZE;
        int end =  current * SystemConstants.DEFAULT_PAGE_SIZE;

        String key = SHOP_GEO_KEY + typeId;
        //查询redis  按照距离排序，分页；结果：shopId，distance

//      GEOSEARCH key BYLONLAT(坐标) x y  BYRADIUS(半径) 10 WITHDISTANCE()距离
        GeoResults<RedisGeoCommands.GeoLocation<String>> results = stringRedisTemplate.opsForGeo()
                .search(key,
                        GeoReference.fromCoordinate(x, y),   // 坐标
                        new Distance(5000),   //距离 m
                        // 分页 limit(end);只能返回 0 ~ end
                        RedisGeoCommands.GeoSearchCommandArgs.newGeoSearchArgs().includeDistance().limit(end)
                );

        if (results == null){
            return Result.ok();
        }
        //解析出id
        List<GeoResult<RedisGeoCommands.GeoLocation<String>>> list = results.getContent();
        int size = list.size();
        List<Long> ids = new ArrayList<>(size);
        Map<String,Distance> map = new HashMap<>(size);

        if(size <= from){ // 没有下一页了，直接返回
            return Result.ok();
        }
        // 逻辑分页 获取 from ~ end 部分的数据
        // 第一页 from = 0 ;end = 5 [0,5)
        // 第二页 from = 5 ;end = 10 [5,10)
        list.stream().skip(from)
                .forEach(result ->{
                    // 获取 shopId
                    String shopId = result.getContent().getName();
                    ids.add(Long.valueOf(shopId));
                    // 获取 distance 距离
                    @NonNull Distance distance = result.getDistance();
                    map.put(shopId,distance);
                });

        //根据id查询shop
        String sids = StrUtil.join(",", ids);
        List<Shop> shops = query()
                    .in("id", ids)
                    .last("ORDER BY FIELD(id," + sids + ")")
                    .list();

        for (Shop shop : shops) {
            double distance = map.get(shop.getId().toString()).getValue();
            // 给店铺设置距离
            shop.setDistance(distance);
        }
        //返回
        return Result.ok(shops);
    }
}
